package org.zh.client.tra_aop;

import com.fasterxml.jackson.core.JsonProcessingException;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.lang.reflect.Method;
import java.net.URI;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.sql.Connection;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;

import static org.zh.client.config.GlobalStatic.*;
import static org.zh.client.config.ZHOpenInit.*;
import static org.zh.util.CommonStatic.*;

//@SuppressWarnings("ALL")
@Component
@Aspect
public class ZHOpenTraAdvice {

    private final DataSource dataSource;
    private final SQLiteJDBC sqLiteJDBC;
    private final GlobalTransactionService globalTransactionService;


    public ZHOpenTraAdvice(@Nullable DataSource dataSource, SQLiteJDBC sqLiteJDBC, GlobalTransactionService globalTransactionService) {
        this.dataSource = dataSource;
        this.sqLiteJDBC = sqLiteJDBC;
        this.globalTransactionService = globalTransactionService;
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                zhOpenTransactionTask(Integer.parseInt(System.getProperty(zhOpenTransactionTaskKeepTime)));
            }
        }, Long.parseLong(System.getProperty(zhOpenTransactionTaskDelay)), Long.parseLong(System.getProperty(zhOpenTransactionTaskPeriod)));
    }

    public void zhOpenTransactionTask(int keepTime) {
        List<RemoteTransactionBean> commitBean = new CopyOnWriteArrayList<>();
        synchronized (globalRegistryCache) {
            Instant now = Instant.now();
            globalRegistryCache.entrySet()
                    .stream()
                    .parallel()
                    .flatMap(en -> en.getValue().stream())
                    .filter(rtb -> rtb.getCreateTime().isBefore(now.minus(Duration.ofSeconds(keepTime))))
                    .map(RemoteTransactionBean::getGlobal_TX_ID)
                    .distinct()
                    .forEach(key -> {
                        List<RemoteTransactionBean> list = globalRegistryCache.get(key);
                        List<RemoteTransactionBean> starts = list.stream().filter(rtb -> rtb.getNodeStatus() == NodeStatus.start.value()).toList();
                        List<RemoteTransactionBean> ends = list.stream().filter(rtb -> rtb.getNodeStatus() == NodeStatus.end.value()).toList();
                        if (starts.size() == ends.size()) {
                            commitBean.addAll(ends);
                            globalRegistryCache.remove(key);
                        }
                    });
        }
        Collection<Callable<HttpResponse<String>>> tasks = new ArrayList<>();
        commitBean.stream().filter(RemoteTransactionBean::isCanRollback).forEach(db -> {
            db.setCommit(true);
            tasks.add(() -> {
                HttpResponse<String> response = smartHttpPUT(URI.create(db.getNodeURI() + GLOBAL_ROLLBACK_COMMIT), HttpRequest.BodyPublishers.ofString(om.writeValueAsString(db)), HttpResponse.BodyHandlers.ofString());
                if (response.statusCode() == 200) {
                    sqLiteJDBC.updateRemote(db.getGlobal_TX_ID());
                } else {
                    throw new RuntimeException("远程通知提交 出现错误");
                }
                return response;
            });
        });
        try {
            executorService.invokeAll(tasks);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            executorService.shutdown();
        }
    }

    @Pointcut("@annotation(org.zh.client.tra_aop.ZHOpenGlobalTra)")
    public void pointCutTX() {
    }

    @Around("pointCutTX()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        ZHOpenGlobalTra zt = method.getAnnotation(ZHOpenGlobalTra.class);
        String localBasePath = System.getProperty(httpBasePath);
        String gtString = GLOBAL_ThreadLocal.get();
        boolean isInherit = gtString == null; //是否  是新开启的节点
        boolean isReadOnly = zt.readOnly();
        boolean hasDataSource = dataSource != null;
        boolean localConn = zt.localConn();
        GlobalType typeAnno = zt.globalType();
        Connection connection = null;
        Object proceed = null;
        try {
            if (hasDataSource) {
                // 有数据源的情况
                if (localConn) {
                    connection = dataSource.getConnection();
                    connectionThreadLocal.set(connection);
                    if (isReadOnly) {
                        if (!typeAnno.equals(GlobalType.NONE)) {
                            if (isInherit) {
                                isInheritStarter(localBasePath, false);
                            } else {
                                noInheritStarter(localBasePath, gtString, false);
                            }
                        }
                        connection.setReadOnly(true);
                        proceed = joinPoint.proceed();
                    } else {
                        if (typeAnno.equals(GlobalType.NONE)) {
                            connection.setAutoCommit(false);
                            connection.setTransactionIsolation(zt.localIsolation().value());
                            proceed = joinPoint.proceed();
                            connection.commit();
                            connection.setAutoCommit(true);
                            connection.close();
                        }
                        if (typeAnno.equals(GlobalType.TX)) {
                            if (isInherit) {
                                RemoteTransactionBean rtb = isInheritStarter(localBasePath, true);
                                LocalTransactionCache.put(rtb.getGlobal_TX_ID(), new LocalTransaction(connection, GlobalType.TX, localBasePath, null, null));
                                sqLiteJDBC.insertTX(rtb.getGlobal_TX_ID(), method.getName(), method.getReturnType().getName(), joinPoint.getArgs());
                            } else {
                                RemoteTransactionBean rtb = noInheritStarter(localBasePath, gtString, true);
                                LocalTransactionCache.put(rtb.getGlobal_TX_ID(), new LocalTransaction(connection, GlobalType.TX, rtb.getROOT_URL(), null, null));
                                sqLiteJDBC.insertTX(rtb.getGlobal_TX_ID(), method.getName(), method.getReturnType().getName(), joinPoint.getArgs());
                            }
                            connection.setAutoCommit(false);
                            connection.setTransactionIsolation(zt.localIsolation().value());
                            proceed = joinPoint.proceed();
                        }
                        if (typeAnno.equals(GlobalType.TA)) {
                            if (zt.taSql().isEmpty()) {
                                throw new RuntimeException("TA全局事务中  回滚sql 不能为空");
                            }
                            if (zt.sqlParamIndex() < 0) {
                                throw new RuntimeException("TA全局事务中  回滚sql参数必须是你方法参数的下坐标从1开始  只支持Map array Collection 三种类型");
                            }
                            Object arg = joinPoint.getArgs()[zt.sqlParamIndex() - 1];
                            if (isInherit) {
                                RemoteTransactionBean rtb = isInheritStarter(localBasePath, true);
                                LocalTransactionCache.put(rtb.getGlobal_TX_ID(), new LocalTransaction(null, GlobalType.TA, localBasePath, zt.taSql(), arg));
                                sqLiteJDBC.insertTA(rtb.getGlobal_TX_ID(), localBasePath, zt.taSql(), arg);
                            } else {
                                RemoteTransactionBean rtb = noInheritStarter(localBasePath, gtString, true);
                                LocalTransactionCache.put(rtb.getGlobal_TX_ID(), new LocalTransaction(null, GlobalType.TA, rtb.getROOT_URL(), zt.taSql(), arg));
                                sqLiteJDBC.insertTA(rtb.getGlobal_TX_ID(), rtb.getROOT_URL(), zt.taSql(), arg);
                            }
                            connection.setAutoCommit(false);
                            connection.setTransactionIsolation(zt.localIsolation().value());
                            proceed = joinPoint.proceed();
                            connection.commit();
                            connection.setAutoCommit(true);
                            connection.close();
                        }
                    }
                }
            }
            if ((!hasDataSource && !typeAnno.equals(GlobalType.NONE)) || (!localConn && !typeAnno.equals(GlobalType.NONE))) {
                if (isInherit) {
                    isInheritStarter(localBasePath, false);
                } else {
                    noInheritStarter(localBasePath, gtString, false);
                }
                proceed = joinPoint.proceed();
            }
        } catch (Exception e) {
            if (connection != null) {
                if (!connection.getAutoCommit()) {
                    connection.rollback();
                    connection.setAutoCommit(true);
                }
                connection.close();
            }
            if (!typeAnno.equals(GlobalType.NONE)) {
                String errorRTB = GLOBAL_ThreadLocal.get();
                Thread.startVirtualThread(() -> {
                    try {
                        RemoteTransactionBean rtb = om.readValue(errorRTB, RemoteTransactionBean.class);
                        rtb.setNodeStatus(NodeStatus.error.value());
                        HttpResponse<String> response = smartHttpPUT(URI.create(rtb.getROOT_URL() + GLOBAL_ERROR), HttpRequest.BodyPublishers.ofString(om.writeValueAsString(rtb)), HttpResponse.BodyHandlers.ofString());
                        if (response.statusCode() != 200) {
                            globalTransactionService.commitOrRollback(rtb.getGlobal_TX_ID(), false);
                            throw new RuntimeException("错误信息注册失败" + response.statusCode() + "::" + response.body());
                        }
                    } catch (JsonProcessingException ex) {
                        throw new RuntimeException(ex);
                    }
                });
            }
            typeAnno = GlobalType.NONE;
            throw new RuntimeException("不知道为什么报错了", e);
        } finally {
            connectionThreadLocal.remove();
            if (!typeAnno.equals(GlobalType.NONE)) {
                String endRTB = GLOBAL_ThreadLocal.get();
                Thread.startVirtualThread(() -> {
                    try {
                        RemoteTransactionBean rtb = om.readValue(endRTB, RemoteTransactionBean.class);
                        rtb.setNodeStatus(NodeStatus.end.value());
                        HttpResponse<String> response = smartHttpPUT(URI.create(rtb.getROOT_URL() + GLOBAL_REGISTRY), HttpRequest.BodyPublishers.ofString(om.writeValueAsString(rtb)), HttpResponse.BodyHandlers.ofString());
                        if (response.statusCode() != 200) {
                            globalTransactionService.commitOrRollback(rtb.getGlobal_TX_ID(), false);
                            throw new RuntimeException("方法完成注册失败" + response.statusCode() + "::" + response.body());
                        }
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                });
            }
        }
        return proceed;
    }

    private RemoteTransactionBean isInheritStarter(String localBasePath, boolean isCanRollback) throws JsonProcessingException {
        String global_tx_id = getGlobal_TX_ID();
        RemoteTransactionBean rtb = new RemoteTransactionBean(global_tx_id, localBasePath, isCanRollback, localBasePath, NodeStatus.start.value());
        Thread.startVirtualThread(() -> {
            try {
                HttpResponse<String> response = smartHttpPUT(URI.create(rtb.getROOT_URL() + GLOBAL_REGISTRY), HttpRequest.BodyPublishers.ofString(om.writeValueAsString(rtb)), HttpResponse.BodyHandlers.ofString());
                if (response.statusCode() != 200) {
                    globalTransactionService.commitOrRollback(rtb.getGlobal_TX_ID(), false);
                    throw new RuntimeException("远程注册失败::" + response.statusCode() + "::" + response.body());
                }
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
        });
        GLOBAL_ThreadLocal.set(om.writeValueAsString(rtb));
        return rtb;
    }

    private RemoteTransactionBean noInheritStarter(String localBasePath, String gtString, boolean isCanRollback) {
        try {
            RemoteTransactionBean rtb = om.readValue(gtString, RemoteTransactionBean.class);
            RemoteTransactionBean localNode = new RemoteTransactionBean(rtb.getGlobal_TX_ID(), rtb.getROOT_URL(), isCanRollback, localBasePath, NodeStatus.start.value());
            Thread.startVirtualThread(() -> {
                try {
                    HttpResponse<String> response = smartHttpPUT(URI.create(rtb.getROOT_URL() + GLOBAL_REGISTRY), HttpRequest.BodyPublishers.ofString(om.writeValueAsString(localNode)), HttpResponse.BodyHandlers.ofString());
                    if (response.statusCode() != 200) {
                        globalTransactionService.commitOrRollback(rtb.getGlobal_TX_ID(), false);
                        throw new RuntimeException("远程注册失败" + response.statusCode() + "::" + response.body());
                    }
                } catch (JsonProcessingException e) {
                    throw new RuntimeException(e);
                }
            });
            GLOBAL_ThreadLocal.set(om.writeValueAsString(localNode));
            return localNode;
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }


}